package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2024/03/02/15:37
 * @Description: 这个类统一管理内存中的数据
 * 该类提供的方法,可能会在多线程环境下使用,考虑线程安全问题
 */
public class MemoryDataCenter {
    //交换机   key是exchangeName value是exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //队列   key是queueName,value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //绑定  第一个key是exchangeName,第二个key是queueName,value是binding对象
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //消息   key是messageId,value是message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //队列和消息之间的关联     key是queueName,value是message链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //未被确认的消息 key是queueName,value是hash表,value的key是messageId,value是message对象
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();


     //插入交换机
    public void insertExchange(Exchange exchange) {
        //将交换机对象插入到上述的hash表中
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName=" + exchange.getName());
    }

    //获取交换机
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    //删除交换机
    public void deleteExchange(String  exchangeName) {
        //将交换机从上述的hash表中删除
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName);

    }


    //针对队列的方法
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功! queueName=" + queue.getName());
    }
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName=" + queueName);
    }

    //针对绑定的方法
    //插入绑定
    public void insertBinding(Binding binding) throws MqException {
        //先使用exchangeName查一下是否存在,不存在就创建
//        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if(bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(),bindingMap);
//        }
        //先使用exchangeName查一下是否存在,不存在就创建
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        //针对数据进行进一步插入
        //再根据queueName查询,如果已经存在,就抛出异常,不存在就插入
        synchronized (bindingMap) {
            if(bindingMap.get(binding.getQueueName())!= null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName=" +
                        binding.getExchangeName() + ",queueName = "+ binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(),binding);
        }
        System.out.println("[MemoryDataCenter] 新的绑定添加成功! exchangeName=" + binding.getExchangeName()+
                ",queueName=" + binding.getQueueName());

    }

    //获取绑定 写两个版本
    //1. 根据exchangeName和queueName确定唯一的Binding
    //2. 根据exchangeName 获取所有的Binding
    // 版本1 确定唯一的Binding
    public Binding getBinding(String exchangeName,String queueName) {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }
    //版本2 获取所有的Binding
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    //删除绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null) {
            //该交换机没有绑定任何队列
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName = " + binding.getExchangeName()
            +",queueName="+ binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName=" + binding.getExchangeName()+
                ",queueName=" + binding.getQueueName());

    }

    //消息管理
    //添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功! messageId=" + message.getMessageId());
    }

    //根据messageId获取消息
    public Message getMessage(String MessageId) {
        return messageMap.get(MessageId);
    }

    //根据messageId删除消息
    public  void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除! messageId = " + messageId);
    }

    //发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message) {
        //就是把消息放在队列对应的数据结构中
        //根据队列中名称找到消息链表
        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
        if(messages == null) {
            messages = new LinkedList<>();
            queueMessageMap.put(queue.getName(),messages);
        }
        //再把数据放到messages里面  确保线程安全
        synchronized (messages) {
            messages.add(message);
        }
        //这里把该消息也往总的消息中心插入一下
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中!  messageId= " + message.getMessageId());
    }

    //从队列中取消息
    public Message pollMessage(String queueName) {
        //根据队列名查询一下队列对应的消息链表
        //如果没有找到,说明队列中没有任何消息
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null) {
            return null;
        }
        //链表中有元素,就进行头删除
        synchronized (messages) {
            if(messages.size() == 0) {
                return null;
            }
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出!  messageId= " + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    //获取指定队列中消息的个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null) {
            //队列中没有消息
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    //未确认消息的处理
    //添加未确认的消息
    public void addMessageWaitAck(String queueName,Message message) {
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列!  messageId= " + message.getMessageId());
    }
    //删除未确认的消息(消息已经确认)
    public void removeMessageWaitAck(String queueName,String messageId) {
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null) {
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除!  messageId= " + messageId);

    }
    //获取指定未确认的消息
    public Message getMessageWaitAck(String queueName,String messageId) {
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null) {
            return null;
        }
        return messageHashMap.get(messageId);
    }


    //这个方法就是从硬盘上读取数据, 把硬盘中持久化存储的各个维度的数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0 清除所有旧的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();

        //1.恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for(Exchange exchange:exchanges) {
            exchangeMap.put(exchange.getName(),exchange);
        }
        //2.恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueue();
       for(MSGQueue queue: queues) {
           queueMap.put(queue.getName(),queue);
       }
        //3.恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
       for(Binding binding : bindings) {
           ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                   k-> new ConcurrentHashMap<>());
           bindingMap.put(binding.getQueueName(),binding);
       }
        //4.恢复所有的消息数据
        //遍历所有的队列 再去根据每个队列的名字取获取到所有的消息
        for(MSGQueue queue: queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for(Message message: messages) {
                messageMap.put(message.getMessageId(),message);
            }
        }
        //针对未确认的消息 这部分内存中的数据,不需要从硬盘恢复
        //在等待ACK的过程中,服务器重启了,此时未被确认的消息 就 恢复成未被取走的消息
        //这个消息在硬盘上存储的时候,就是当做 未被取走
    }
}
